package com.yifeng.repo.controller.kafka.message.consume;

import com.talkyun.utils.revised.Looper;
import com.yifeng.repo.base.constant.BaseConstant;
import com.yifeng.repo.controller.kafka.message.handler.KafkaHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.function.Consumer;

/**
 * @author 00115764
 * Created by daibing on 2024/5/2.
 */
@Slf4j
public class SubscribePullConsumerNoPausedLooper extends Looper {
    private final KafkaConsumer<String, String> pullConsumer;
    private final Consumer<ConsumerRecords<String, String>> msgListConsumer;
    private long lastPrintTime = 0L;

    public SubscribePullConsumerNoPausedLooper(String name, int interval, int errorInterval,
                                               KafkaConsumer<String, String> pullConsumer,
                                               Consumer<ConsumerRecords<String, String>> msgListConsumer) {
        super(name, interval, errorInterval);
        this.pullConsumer = pullConsumer;
        this.msgListConsumer = msgListConsumer;
        log.info("SubscribePullConsumerLooper init: name={}, partitions={}",
                name, KafkaHelper.buildPartitionLog(pullConsumer.assignment()));
    }

    @Override
    protected void loop() throws Throwable {
        // 1.拉取&消费消息
        ConsumerRecords<String, String> consumerRecords = pullConsumer.poll(Duration.ofMillis(5 * 1000));
        msgListConsumer.accept(consumerRecords);

        // 2. 间歇性打印消费分区信息
        if (lastPrintTime + BaseConstant.ONE_HOUR_MS < System.currentTimeMillis()) {
            log.info("SubscribePullConsumerLooper ok: name={}, assignment={}, size={}",
                    getName(), KafkaHelper.buildPartitionLog(pullConsumer.assignment()), consumerRecords.count());
            lastPrintTime = System.currentTimeMillis();
        }
    }

    @Override
    protected void loopThrowable(Throwable t) {
        log.warn("consumer {} failed: error=", getName(), t);
    }

    @Override
    protected void clear() {
        pullConsumer.unsubscribe();
        pullConsumer.close();
    }

    public KafkaConsumer<String, String> getPullConsumer() {
        return pullConsumer;
    }

}
